package com.biglead.orderservice.service.impl;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.biglead.common.config.Constant;
import com.biglead.common.pojo.GoodsInfo;
import com.biglead.common.pojo.OrderInfo;
import com.biglead.common.pojo.OrderMessage;
import com.biglead.orderservice.mapper.OrderMapper;
import com.biglead.orderservice.service.GoodsInfoService;
import com.biglead.orderservice.service.OrderKafakaService;
import com.biglead.orderservice.service.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Date;

@Service
@Slf4j
public class OrderKafakaServiceImpl extends ServiceImpl<OrderMapper, OrderInfo> implements OrderKafakaService {

    @Resource
    private GoodsInfoService goodsInfoService;

    @Resource
    RedisService redisService;

    @Override
    @Transactional
    public void consumerTopicToCreateOrderWithKafka(OrderMessage orderMessage) throws Exception {
        Long goodsId = orderMessage.getGoodsId();
        Long userId = orderMessage.getUserId();
        String sn = orderMessage.getSn();
        Integer num = orderMessage.getNum();
        // 第一步 校验库存，从 Redis 中获取
        GoodsInfo goodsInfo = checkStockWithRedis(goodsId);
        if(goodsInfo==null){
            //更新下单标识下单失败
            redisService.set(Constant.ORDER_SN+sn,3);
            return;
        }
        // 第二步 乐观锁更新库存和Redis
        boolean update = saleStockOptimsticWithRedis(goodsInfo);
        if(!update){
            //更新下单标识下单失败
            redisService.set(Constant.ORDER_SN+sn,3);
            return;
        }
        //第三步 创建订单
        GoodsInfo goodsInfo1 = (GoodsInfo) redisService.get(Constant.GOODS_INFO + goodsId);

        OrderInfo order = new OrderInfo();

        order.setSn(sn);
        order.setGoodsId(goodsId);
        order.setUserId(userId);
        order.setNum(num);
        order.setPrice(goodsInfo1.getGoodsPrice().multiply(new BigDecimal(num)));
        order.setName(goodsInfo1.getGoodsName());
        order.setCreateTime(new Date());
        order.setOrderStatues(0);
        boolean res = this.save(order);
        if (!res) {
            //更新下单标识下单失败
            redisService.set(Constant.ORDER_SN+sn,3);
            return;

        }

        //第四步 更新预下订单标识
        redisService.set(Constant.ORDER_SN+sn,2);
        //未支付取消订单 设置过期时间30分钟
        redisService.expire(Constant.ORDER_SN+sn,60*30);
        //缓存订单信息
        redisService.set(Constant.ORDER_SN_NORMAL+sn,order);

        log.info("Kafka 消费 Topic 创建订单成功");


    }

    // Redis 中校验库存
    private GoodsInfo checkStockWithRedis(Long goodsId) throws Exception {
        int count = Integer.parseInt(redisService.get(Constant.GOOOD_STOCK + goodsId).toString());
        int sale = Integer.parseInt(redisService.get(Constant.GOOOD_SALE + goodsId).toString());
        int version = Integer.parseInt(redisService.get(Constant.GOOOD_VERSION + goodsId).toString());
        String goodsName = redisService.get(Constant.GOODS_NAME + goodsId).toString();
        if (count < 1) {
            log.info("库存不足");
            return null;
        }
        GoodsInfo stock = new GoodsInfo();
        stock.setId(goodsId);
        stock.setGoodsStock(count);
        stock.setGoodsSale(sale);
        stock.setVersion(version);
        stock.setGoodsName(goodsName);

        return stock;
    }


    // 更新 DB 和 Redis
    private boolean saleStockOptimsticWithRedis(GoodsInfo goodsInfo) throws Exception {
        boolean res = goodsInfoService.updateStockByOptimistic(goodsInfo);
        if (!res) {
           return false;
        }
        // 更新 Redis
        redisService.updateStockWithRedis(goodsInfo);
        return true;
    }

}
